package com.dd.ms.mw.kafka.utils;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @Author liuxianmeng
 * @CreateTime 2025/6/16 22:56
 * @Description Kafka生产者工具类 用来发送Kafka消息
 */
@Component
public class KafkaProducerUtil {

    private static final org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(KafkaProducerUtil.class);

    //@Value("${log.system.topic}")
    public static final String systemLogTopic = "system_log";
    
    private static KafkaTemplate<String, Object> kafkaObjectTemplate;
    
    public KafkaProducerUtil(KafkaTemplate<String, Object> kafkaObjectTemplate) {
        KafkaProducerUtil.kafkaObjectTemplate = kafkaObjectTemplate;
    }

    /**
     * 通用消息发送方法（带自定义回调）
     *
     * @param topic 主题名称
     * @param data 消息内容
     * @param callback 自定义回调处理器
     */
    public static void sendMessage(String topic, Object data, SendCallback callback) {
        ListenableFuture<SendResult<String, Object>> future = kafkaObjectTemplate.send(topic, data);
        future.addCallback(callback);
    }

    /**
     * 通用消息发送方法（带默认回调）
     *
     * @param topic 主题名称
     * @param data 消息内容
     */
    public static void sendMessage(String topic, Object data) {
        try {
            sendMessage(topic, data, new DefaultSendCallback());
        } catch (Exception e) {
            log.error("日志发送失败：" + e.getMessage(), e);
        }
    }

    // 回调接口定义
    public interface SendCallback extends ListenableFutureCallback<SendResult<String, Object>> {
    }

    // 默认回调实现
    private static class DefaultSendCallback implements SendCallback {
        @Override
        public void onSuccess(SendResult<String, Object> result) {
            assert result != null;
            log.info("Message sent successfully! Topic: {}, Partition: {}, Offset: {}",
                    result.getRecordMetadata().topic(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
        }

        @Override
        public void onFailure(Throwable ex) {
            log.error("Message sending failed! Reason: {}", ex.getMessage(), ex);
            // 这里可以添加重试逻辑或告警通知
        }
    }
}
